package extractor;

import java.io.File;
import java.io.IOException;
import java.net.URI;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;

import net.imdict.analysis.chinese.AnalyzerProfile;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.SequenceFileInputFormat;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.lib.MultipleInputs;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import preprocess.AttachSalerShopToItem;
import utils.Tokenizer;
import utils.Utils;

public class AuctionItemExtractor extends Configured implements Tool{
	public static String getPvItemDetailPath(String date) {
		//return "D:\\dev\\clickprediction\\data\\PvItemDetail";
		return Utils.rootDir + date + "/PvItemDetail/";
	}
	enum Counters{AUCTION_53, AUCTION_34, INVALID_AUCTION, INVALID_PV_CLICK, INVALID_ITEM_SALER_SHOP,  
		PV_CLICK_JOINED_IN_CORRECTLY, CLICKED_ITEM, UNCLICKED_ITEM,
		AUCTION_REPLICATED, ITEM_SALER_SHOP_REPLICATED, AUCTION_NOT_FOUND, AUCTION_FOUND, PV_ITEM_JOINED, INVALID_REDUCE_ITEM,
		ITEM_SALER_SHOP_NOT_FOUND, ITEM_SALER_SHOP_FOUND, CATEGORY_NOTFOUND};
	private static final String CLICK = "CLK";
	private static final String NOT_CLICK = "NCLK";
	private static final String CATEGORY_PATH_NAME = "CATEGORY_PATH_NAME";
	private static final String CATEGORY_COEC_PATH_NAME = "CATEGORY_COEC_PATH_NAME";
	private static final String ROOT_CATEGORY_COEC_PATH_NAME = "ROOT_CATEGORY_COEC_PATH_NAME";
	
	public static class AuctionMap extends MapReduceBase implements Mapper<BytesWritable, Text, Text, Text> {
		private Text k = new Text();
		private Text v = new Text();
		
		public void map(BytesWritable key, Text value, 
				OutputCollector<Text, Text> output, Reporter reporter) throws IOException {
			
			String[] values = value.toString().split(Utils.auctionSeparator);
			if (values.length == 53) {
				reporter.incrCounter(Counters.AUCTION_53, 1);
				
			} else if (values.length == 34) {
				reporter.incrCounter(Counters.AUCTION_34, 1);
			} else if (values.length < 34){
				reporter.incrCounter(Counters.INVALID_AUCTION, 1);
				return;
			}
			
			k.set(values[18]);
			v.set(Utils.getRecord(values, Utils.separator, 0, 34));
			output.collect(k, v);
		}
	}
	public static class PvClickMap extends MapReduceBase implements Mapper<LongWritable, Text, Text, Text> {
		private Text k = new Text();
		private Text v = new Text();
		private final int[] searchColumns = new int[]{1, 2, 4, 5, 8, 9, 10, 12, 13, 14, 15, 16, 17, 18, 25};
		
		public void map(LongWritable key, Text value, 
				OutputCollector<Text, Text> output, Reporter reporter) throws IOException {

			String[] values = value.toString().split(Utils.separator);
			if (values.length != 61) {
				reporter.incrCounter(Counters.INVALID_PV_CLICK, 1);
				return;
			}
			
			
			HashMap<String, String[]> clickItemMap = new HashMap<String, String[]>();
			if (Utils.isValid(values[58])) {
				String[] clicks = values[58].split("\u0002");
				for(String click : clicks) {
					if (!Utils.isValid(click))
						continue;
					String[] clickInfo = click.split("\u0001");
					clickItemMap.put(clickInfo[3], clickInfo);
				}
			}
			
			String[] itemsTypes = values[28].split(Utils.pvAuctionSeparator1);
			int pos = 0;
			for (int i = 0; i < itemsTypes.length; ++i) {
				String itemString = itemsTypes[i];
				int isDoufu = 0;
				if (itemsTypes.length == 2 && i == 0) {
					isDoufu = 1;
				}
				if (!Utils.isValid(itemString))
					continue;				
				
				String[] items = itemString.split(Utils.pvAuctionSeparator2);
				for (String item : items) {
					++pos;
					//点击时间^A搜索的来源pid^A点击位置^A点击商品id(数字)^A是否为豆腐块商品
					String[] clickInfo = null;
					String isClicked = NOT_CLICK;
					if (clickItemMap.containsKey(item.toString())) {
						isClicked = CLICK;
						clickInfo = clickItemMap.get(item.toString());	
						if (pos != Integer.parseInt(clickInfo[2])
								|| isDoufu != Integer.parseInt(clickInfo[4])) {
							reporter.incrCounter(Counters.PV_CLICK_JOINED_IN_CORRECTLY, 1);
							//k.set("PV_CLICK_JOINED_IN_CORRECTLY");				
							//output.collect(k, value);
							//return;
							continue;
						}
						reporter.incrCounter(Counters.CLICKED_ITEM, 1);
						
					} else {						
						clickInfo = new String[]{"", "", "", "", ""};
						clickInfo[2] = String.valueOf(pos);
						clickInfo[3] = item;
						clickInfo[4] = String.valueOf(isDoufu);
						reporter.incrCounter(Counters.UNCLICKED_ITEM, 1);
					}
					String[] newValues = new String[43];
					

					int index = 0;
					//search
					for (int j = 0; j < searchColumns.length; ++j, ++index) {
						newValues[index] = values[0 + searchColumns[j] - 1];
					}

					//pv
					newValues[index++] = values[25];//uuid
					newValues[index++] = values[26];//pv时间
					newValues[index++] = values[27];//直达后台类目
					newValues[index++] = isClicked;//是否点击
					//buyer
					final int[] userColumns = UserExtendedExtractor.getUserColumns();
					for (int j = 0; j < userColumns.length; ++j, ++index) {
						newValues[index] = values[29 + userColumns[j] - 1];
					}
					//click
					for (int j = 0; j < clickInfo.length; ++j, ++index)
						newValues[index] = clickInfo[j];
					newValues[index++] = values[59];
					newValues[index++] = values[60];
					//item			
						                      
					k.set(item);
					v.set(Utils.getRecord(newValues, Utils.separator));
					output.collect(k, v);
				}
			}
		}		
	}
	
	public static class ItemSalerShopMap extends MapReduceBase implements Mapper<LongWritable, Text, Text, Text> {
		private Text k = new Text();
		private Text v = new Text();
		@Override
		public void map(LongWritable key, Text value, 
				OutputCollector<Text, Text> output, Reporter reporter) throws IOException {
			// TODO Auto-generated method stub
			String[] values = value.toString().split(Utils.separator);
			if (values.length != 47) {
				reporter.incrCounter(Counters.INVALID_ITEM_SALER_SHOP, 1);
				return;
			}
			
			k.set(values[45]);
			v.set(value.toString());
			output.collect(k, v);
		}
		
	}
	
	
	
	public static class Reduce extends MapReduceBase implements Reducer<Text, Text, Text, Text> {
		private Text k = new Text();
		private Text v = new Text();
		
		private Tokenizer tokenizer = null;
	    private CategoryHelper categoryHelper = new CategoryHelper();
	    private CategoryCoecHelper cateCoecHelper = new CategoryCoecHelper();
	    private CategoryCoecHelper rootCateCoecHelper = new CategoryCoecHelper();
		@Override   
	    public void configure(JobConf job) {
	         // Get the cached archives/files
	         try {
	        	 Path[] localArchives = DistributedCache.getLocalCacheArchives(job);
	        	 File file = new File(localArchives[0].toString());
	        	 AnalyzerProfile.init(file.getAbsolutePath() + "/analysis-data");
	        	 System.out.println("AnalyzerProfile dir: " + AnalyzerProfile.ANALYSIS_DATA_DIR);
	        	 tokenizer = new Tokenizer();
	        	 String categoryPath = job.get(CATEGORY_PATH_NAME);
	        	 String cateCoecPath = job.get(CATEGORY_COEC_PATH_NAME);
	        	 String rootCateCoecPath = job.get(ROOT_CATEGORY_COEC_PATH_NAME);
	        	 categoryHelper.init(FileSystem.get(job), new Path(categoryPath), job);
	        	 cateCoecHelper.init(FileSystem.get(job), new Path(cateCoecPath), job);
	        	 rootCateCoecHelper.init(FileSystem.get(job), new Path(rootCateCoecPath), job);
	        	 
			} catch (IOException e) {
				// TODO Auto-generated catch block
				System.out.println(e.toString());
			}
	    }
		
		@Override
		public void reduce(Text key, Iterator<Text> iter, OutputCollector<Text, Text> output, 
				Reporter reporter) throws IOException {
			
			ArrayList<String[]> pvClickList = new ArrayList<String[]>();
			String[] values = null;
			String[] auctionValues = null;
			
			int auctionCount = 0;
			int itemSalerCount = 0;
			String salerShop = null;
			while (iter.hasNext()) {
				String value = iter.next().toString();
				values = value.split(Utils.separator);
				int columnsCount = values.length;
				if (columnsCount == 34) {
					if (auctionCount != 0) {
						reporter.incrCounter(Counters.AUCTION_REPLICATED, 1);
						continue;
					}
					auctionCount = 1;
					auctionValues = values;					
				} else if (columnsCount == 43) {
					pvClickList.add(values);
				} else if (columnsCount == 47) {
					if (itemSalerCount != 0) {
						reporter.incrCounter(Counters.ITEM_SALER_SHOP_REPLICATED, 1);
						continue;
					}
					itemSalerCount = 1;
					salerShop = Utils.getRecord(values, Utils.separator, 0, 44);	
					
				} else {
					reporter.incrCounter(Counters.INVALID_REDUCE_ITEM, 1);
					continue;
				}
			}
			if (pvClickList.size() == 0)
				return;
			
			if (auctionValues == null) {
				reporter.incrCounter(Counters.AUCTION_NOT_FOUND, 1);
				return;
			}
			if (salerShop == null) {
				reporter.incrCounter(Counters.ITEM_SALER_SHOP_NOT_FOUND, 1);
				return;
			}
			
			reporter.incrCounter(Counters.ITEM_SALER_SHOP_FOUND, 1);
			reporter.incrCounter(Counters.AUCTION_FOUND, 1);

			auctionValues[6] = tokenizer.tokenizeToString(auctionValues[6]);
			String category = auctionValues[9];
			String cateName, rootCategory, rootCateName;
			
			String[] categoryInfo = categoryHelper.getCategoryInfo(category);
			if (categoryInfo == null) {
				reporter.incrCounter(Counters.CATEGORY_NOTFOUND, 1);
				cateName = rootCategory = rootCateName = "";
			}
			else {
				cateName = categoryInfo[2];
				rootCategory = categoryInfo[9];
				rootCateName = categoryInfo[12];
			}
			String valueString1 = Utils.getRecord(
					new String[]{Utils.getRecord(auctionValues, Utils.separator), 
							salerShop, cateName, rootCategory, rootCateName}, 
					Utils.separator);
			String cateCoec = cateCoecHelper.getCategoryCoecCtrStr(category);
			String rootCateCoec = rootCateCoecHelper.getCategoryCoecCtrStr(rootCategory);
			for (String[] pvClick : pvClickList) {
				pvClick[1] =tokenizer.tokenizeToString(pvClick[1]);
				k.set(Utils.getRecord(pvClick, Utils.separator, 0, 41));
				//buyer coec, ctr
				String valueString2 = Utils.getRecord(pvClick, Utils.separator, 41, 2);
				v.set(Utils.getRecord(new String[]{valueString1, valueString2, cateCoec, rootCateCoec}, Utils.separator));
				reporter.incrCounter(Counters.PV_ITEM_JOINED, 1);
				output.collect(k, v);
			}
		}
		
	}
	
	/*
	 * (non-Javadoc)
	 * @see org.apache.hadoop.util.Tool#run(java.lang.String[])
	 * Path: /group/tbsc-dev/xzwang/clickprediction/$date/PvItemDetail/
	 * ---------------------search---------------------
		1.user 32位id
		2.Query 
		3.User 数字id
		4.用户分类
		5.点击时间
		6.搜索的来源（1：TB站外来源，2：TB站内搜索结果页引流，3：TB站内首页用户主动搜索，4：新版搜索首页用户主动搜索，5：新版搜索相关搜索。现在该字段不一定是数字，有可能是域名）
		7.搜索类目
		8.排序,格式一转换
		9.列表模式（大图、小图）
		10.消保筛选
		11.所在地筛选
		12.结果数 01：违禁词过滤造成无结果 02：超时无结果 03：iSearch真实无结果 0：未知原因无结果
		13.第几页
		14.搜索属性
		15.Tab类型 (0:未知，1：所有宝贝，2：人气，3：淘宝商城，4：二手 
		---------------------pv---------------------
	 	16.UUID
	 	17.pv时间
	 	18.直达后类目
	 	19.是否点击
	 	---------------------buyer---------------------
		20.CITY (所在城市)
		21.PROV (所在省份)
		22.RATE_SUM (作为卖家的信用分值，只有当等级发生变迁的时候才更新)
		23.RATE_NUM (作为买家的信用分值，只有当等级发生变迁的时候才更新)
		24.BIRTHDATE (生日)
		25.SUSPENDED (用户状态：0 正常;1 未激活;2 删除;3 冻结;-9 CC) 
		26.USER_SESSION_TIME (用户激活时间) 
		27.USER_LEVERL (用户等级) 
		28.USER_STYLE (收费服务类型(按位取与字段，目前定义的二进制位有 1:alisoft网店版用户 2:b2c商城用户[占用先]))
		29.USER_RANK (用户社区等级，论坛改造后没有在使用。现在还是有使用的,在商品举报的地方,有个小二专用, 中用user_rank来判断小二身份...小二身份=46 by yekai)
		30.USER_GENDER (性别)
		31.BUYER_GOODNUM (作为买家的好评数)
		32.SELLER_GOODNUM (作为卖家的好评数)
		33.BUYER_SUM　　(作为买家的评价分值)
		34.BUYER_NUM (作为买家的评价次数)
		35.SELLER_SUM (作为卖家的评价分值)
		36.SELLER_NUM (作为卖家的评价次数)
		---------------------click---------------------
		37.点击时间
		38.搜索的来源pid
		39.点击位置
		40.点击商品id(数字)
		41.是否为豆腐块商品
		---------------------item---------------------
		42.item[1~34]
		---------------------saler---------------------
		76.CITY(所在城市)
		77.PROV(所在省份)
		78.RATE_SUM(作为卖家的信用分值，只有当等级发生变迁的时候才更新)
		79.RATE_NUM(作为买家的信用分值，只有当等级发生变迁的时候才更新)
		80.BIRTHDATE(生日)
		81.SUSPENDED(用户状态：0
		82.USER_SESSION_TIME(用户激活时间)
		83.USER_LEVERL(用户等级)
		84.USER_STYLE(收费服务类型(按位取与字段，目前定义的二进制位有
		85.USER_RANK(用户社区等级，论坛改造后没有在使用。现在还是有使用的,在商品举报的地方,有个小二专用,
		86.USER_GENDER(性别)
		87.BUYER_GOODNUM(作为买家的好评数)
		88.SELLER_GOODNUM(作为卖家的好评数)
		89.BUYER_SUM(作为买家的评价分值)
		90.BUYER_NUM(作为买家的评价次数)
		91.SELLER_SUM(作为卖家的评价分值)
		92.SELLER_NUM(作为卖家的评价次数
		---------------------shop--------------------
		93.shop[1~22]
		115. pvCount
		116.clickCount
		117.saler 32位id
	 	118.saler coec
	 	119.saler ctr
		120.商品类目名称
		121.商品根类目id
		122.商品根目录名称
		123.Buyer coec
		124.Buyer ctr
		125.category coec
		126.category ctr
		127.root category coec
		128.root category ctr
	 * 
	 */
	public int run(String[] args) throws Exception {
		//String itemFile = "D:\\dev\\clickprediction\\dt\\20100531\\Auction";

		// input
		String itemFile = "/group/taobao/taobao/hive/r_auction_auctions/pt=" + args[0] + "000000/if_online=1";
		String pvClickFile = AttachBuyerCoec.getBuyerCoecPath(args[0]);
		String itemSalerShopFile = AttachSalerShopToItem.getUniqueItemSalerShopPath(args[0]);
		String categoryFile = "/group/taobao/taobao/hive/r_auction_category/pt=" + args[0] + "000000/*";
		String categoryCoecFile = "/group/tbsc-dev/xzwang/clickprediction/cyl/categoryFeature/coec/*";
		String rootCategoryCoecFile = "/group/tbsc-dev/xzwang/clickprediction/cyl/rootCategoryFeature/coec/*";
		// Output
		String output = getPvItemDetailPath(args[0]);
		System.out.println("start job SplitPvClickItem...");
		System.out.println("itemFile: " + itemFile);
		System.out.println("pvClickFile: " + pvClickFile);
		System.out.println("itemSalerShopFile: " + itemSalerShopFile);
		System.out.println("categoryFile: " + categoryFile);
		System.out.println("categoryCoecFile: " + categoryCoecFile);
		System.out.println("rootCategoryCoecFile: " + rootCategoryCoecFile);
		
		System.out.println("output: " + output);
		if (Utils.exitSystem(args))
			return -1;
		
		JobConf conf = new JobConf(getConf(), AuctionItemExtractor.class);
		conf.set(CATEGORY_PATH_NAME, categoryFile);
		conf.set(CATEGORY_COEC_PATH_NAME, categoryCoecFile);
		conf.set(ROOT_CATEGORY_COEC_PATH_NAME, rootCategoryCoecFile);
		conf.setJobName("AuctionItemExtractor");
		conf.setOutputKeyClass(Text.class);
		conf.setOutputValueClass(Text.class);

		MultipleInputs.addInputPath(conf, 
				new Path(itemFile),
				SequenceFileInputFormat.class, 
				AuctionMap.class);
	
		MultipleInputs.addInputPath(conf, 
				new Path(pvClickFile),
				TextInputFormat.class,
				PvClickMap.class);
		
		MultipleInputs.addInputPath(conf, 
				new Path(itemSalerShopFile),
				TextInputFormat.class,
				ItemSalerShopMap.class);
	
		FileOutputFormat.setOutputPath(conf, new Path(output));
		DistributedCache.addCacheArchive(new URI(Utils.DICT_PATH), conf);
		//DistributedCache.createSymlink(conf);
		conf.setReducerClass(Reduce.class);
		JobClient.runJob(conf);
		return 0;
	}

	public static void main(String[] args) throws Exception {
		Utils.checkArgs(args);
		
		int res = ToolRunner.run(new Configuration(), new AuctionItemExtractor(),
				args);
		System.exit(res);
	}
}


